package me.xhy.java.springboot3.webflux.projectreactor.s3projectreactor;

import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;

import java.util.List;

// 请求重塑
public class S2ReshapeRequest {
  public static void main(String[] args) {
    // 1. 10个数据，按每组3个（缓冲区大小），分次发布
    System.out.println("1. buffer");
    Flux<Integer> range = Flux.range(1, 10);
    range.subscribe(System.out::println);
    Flux<List<Integer>> buffer = range.buffer(3);
    buffer.subscribe(System.out::println);

    // 2. limitRate 限制频率
    /*
    Flux.range(1, 1000) 产生 1000 元素
    绑定订阅关系、请求 100 个数据 （.limitRate(100)）
    在 log 收到第75个数据的后，请求 75 个数据（.limitRate(100) 的 75% 预取策略，当订阅者消耗预期值的 75%后，再请求75%）
    （在日志中搜 request）
     */
    System.out.println("2. limitRate 限制频率");
    Flux.range(1, 100)
        .log()
        // 预取率 或者 负载因子，如果元素够 每次取3个
        .limitRate(10)
        .subscribe(new BaseSubscriber<Integer>() {
          @Override
          protected void hookOnNext(Integer value) {
            super.hookOnNext(value);
            System.out.println("订阅者 hookOnNext：" + value);
          }
        });

  }
}
